package com.ustadzmenjawab.twitter;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

import twitter4j.RateLimitStatus;
import twitter4j.Status;
import twitter4j.Twitter;
import twitter4j.TwitterException;
import twitter4j.TwitterFactory;
import twitter4j.conf.ConfigurationBuilder;

import com.ustadzmenjawab.conf.Configuration;
import com.ustadzmenjawab.db.DBAccess;
import com.ustadzmenjawab.db.DBConnFactory;

public class ReplyCrawler {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		ConfigurationBuilder cb = new ConfigurationBuilder();
		cb.setDebugEnabled(true)
		.setIncludeRTsEnabled(true)
		.setIncludeEntitiesEnabled(true)
		.setOAuthConsumerKey(Configuration.getProperty("REST_OAUTH_CONSUMER_KEY"))
		.setOAuthConsumerSecret(Configuration.getProperty("REST_OAUTH_CONSUMER_SECRET"))
		.setOAuthAccessToken(Configuration.getProperty("REST_OAUTH_ACCESS_TOKEN"))
		.setOAuthAccessTokenSecret(Configuration.getProperty("REST_OAUTH_ACCESS_TOKEN_SECRET"));
		twitter4j.conf.Configuration twconf = cb.build();
		Twitter restAPI = new TwitterFactory(twconf).getInstance();

		try {
			DBAccess db = new DBAccess();

			while(true){
				List<Long> repliesid = getUncrawledReply();
				if(repliesid == null){
					System.out.println("No reply to process. Sleep for 60 sec");
					Thread.sleep(60 * 1000);
					continue;
				}
				System.out.println("To crawl " + repliesid.size() + " replies.");
				for(long replyid : repliesid){
					try{
						RateLimitStatus rateLimit = restAPI.getRateLimitStatus();
						if(rateLimit.getRemainingHits() > 0){

							System.out.println("Get the message being replied by status " + replyid);
							Status beReplied = restAPI.showStatus(replyid);
							
							long convid = db.insertPublicTweet(beReplied);	
							if(!beReplied.isRetweet() && beReplied.getInReplyToStatusId() == -1)
								db.insertConvTweetSummary(convid, beReplied);
							//updateUntruncatedText(beReplied);
							db.insertRetweet(beReplied);
							

						}else{
							System.out.println("Hit limit exceded, sleep for "+rateLimit.getSecondsUntilReset() + " seconds.");
							try{
							Thread.sleep(rateLimit.getSecondsUntilReset() *1000);
							}catch (InterruptedException e) {
								e.printStackTrace();
							}
						}
					}catch(TwitterException ex){
						if(ex.getExceptionCode().equalsIgnoreCase("6b80c41c-1aae0447")){
							updateIsReplyPublic(replyid, false);
						}else{
							System.out.println("Error code " + ex.getExceptionCode() +":"+ex.getErrorMessage());
						}
					}catch(Exception e){
						e.printStackTrace();
					}
				}
			}
		} catch(Exception e){
			e.printStackTrace();
		}

	}

	public static List<Long> getUncrawledReply(){
		String sql = "SELECT in_reply_to_status_id FROM public_tweet pt" +
		" where in_reply_to_status_id > -1  " +
		" AND is_reply_public = 1" +
		" AND NOT EXISTS (SELECT * FROM public_tweet pt2 where pt.in_reply_to_status_id = pt2.statusid)" +
		" ORDER BY pt.created_at DESC";
		PreparedStatement ps ;
		Connection conn = DBConnFactory.getConnection();
		try{
			ps = conn.prepareStatement(sql);
			ResultSet rs = ps.executeQuery();
			List<Long> arr = new ArrayList<Long>();
			while(rs.next()){
				arr.add(rs.getLong(1));
			}
			if(arr.size() > 0){
				return arr;
			}
			return null;
		}catch (SQLException e) {
			e.printStackTrace();	
			return null;
		}  finally{
			DBConnFactory.closeConnection(conn);
		}	


	}

	public static void updateIsReplyPublic(long replyid, boolean ispublic){
		String sql ="UPDATE public_tweet SET is_reply_public = ? WHERE  in_reply_to_status_id = ?";
		PreparedStatement ps ;
		Connection conn = DBConnFactory.getConnection();
		try{
			ps = conn.prepareStatement(sql);
			ps.setBoolean(1, ispublic);
			ps.setLong(2, replyid);
			ps.execute();
		}catch (SQLException e) {
			e.printStackTrace();	
		}  finally{
			DBConnFactory.closeConnection(conn);
		}	
	}

}
